package com.dyna.utils;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.lang.Assert;
import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.dyna.core.exception.CustomException;
import com.dyna.core.utils.FastBeanUtil;
import com.dyna.core.utils.PageUtil;
import com.dyna.domain.req.PageParam;
import com.dyna.domain.resp.PageResult;
import com.dyna.resp.GroupResp;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.*;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.GetIndexResponse;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.BucketOrder;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedLongTerms;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.MaxAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.MinAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.ParsedMax;
import org.elasticsearch.search.aggregations.metrics.ParsedMin;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.xcontent.XContentType;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author zhoucaiwang
 */
@Slf4j
public class EsUtil {

    private final static String INDEX_MSG = "索引不能为空";
    private final static String ES_ID = "esId";
    private final static String UPDATE = "updated";
    private final static String DELETE = "deleted";
    private static final int BATCH_SIZE = 1000; // 根据ES集群性能调整
    private static final TimeValue DEFAULT_TIMEOUT = TimeValue.timeValueMinutes(2);
    private static final Scroll scroll = new Scroll(DEFAULT_TIMEOUT);


    private static RestHighLevelClient restHighLevelClient;

    static {
        EsUtil.restHighLevelClient = SpringUtil.getBean(RestHighLevelClient.class);
    }

    /**
     * 查询所有索引
     */
    public static List<String> getIndices() {
        GetIndexRequest indexRequest = new GetIndexRequest("*");
        try {
            GetIndexResponse getIndexResponse = restHighLevelClient.indices().get(indexRequest, RequestOptions.DEFAULT);
            String[] indices = getIndexResponse.getIndices();
            return Arrays.stream(indices).collect(Collectors.toList());
        } catch (IOException e) {
            throw new CustomException("查询所有索引异常");
        }
    }

    /**
     * 添加索引
     */
    public static Boolean addIndex(String index) {
        Assert.notEmpty(index,INDEX_MSG);
        boolean result = Boolean.FALSE;
        CreateIndexRequest createIndexRequest = new CreateIndexRequest(index);
        try {
            CreateIndexResponse response = restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
            result = response.isAcknowledged();
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        return result;
    }

    /**
     * 删除索引
     */
    public static Boolean delIndex(String index) {
        Assert.notEmpty(index,INDEX_MSG);
        boolean result = Boolean.FALSE;
        DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(index);
        try {
            AcknowledgedResponse acknowledgedResponse = restHighLevelClient.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
            result = acknowledgedResponse.isAcknowledged();
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        return result;
    }

    /**
     * 索引是否存在
     */
    public static Boolean isExist(String index) {
        Assert.notEmpty(index,INDEX_MSG);
        boolean result = Boolean.FALSE;
        GetIndexRequest getIndexRequest = new GetIndexRequest(index);
        try {
            result = restHighLevelClient.indices().exists(getIndexRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        return result;
    }

    /**
     * 初始化索引
     * @param index 索引
     */
    public static void initIndex(String index) {
        Boolean exist = isExist(index);
        if (!exist) {
            addIndex(index);
        }
    }

    /**
     * 添加数据
     */
    public static String addData(String index,Object obj) {
        EsUtil.initIndex(index);
        IndexRequest addRequest = new IndexRequest().index(index);
        JSONObject jsonObject = JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.WRITE_MAP_NULL_FEATURES));
        log.info("\nelasticsearch index: {} add data : {}", index, jsonObject);
        addRequest.source(jsonObject, XContentType.JSON);
        try {
            IndexResponse addResponse = restHighLevelClient.index(addRequest, RequestOptions.DEFAULT);
            return addResponse.getId();
        } catch (Exception e) {
            log.error("添加es数据失败:\n{}", e.getMessage(), e);
            throw new CustomException("添加es数据失败");
        }
    }

    /**
     * 批量添加数据
     */
    public static <T> void addBatchData(String index,List<T> data) {
        Assert.notBlank(index,INDEX_MSG);
        Assert.notEmpty(data,"新增es数据不能为空");
// 分批处理
        List<List<T>> partitions = Lists.partition(data,  BATCH_SIZE);
        AtomicInteger successCount = new AtomicInteger();

        partitions.parallelStream().forEach(batch  -> {
            BulkRequest bulkRequest = new BulkRequest()
                    .timeout(DEFAULT_TIMEOUT);

            // 并行化文档转换
            List<IndexRequest> indexRequests = batch.parallelStream()
                    .map(item -> new IndexRequest(index)
                            .source(FastBeanUtil.beanToMapFast(item)))
                    .collect(Collectors.toList());

            indexRequests.forEach(bulkRequest::add);

            try {
                // 执行批量请求
                BulkResponse response = restHighLevelClient.bulk(bulkRequest,  RequestOptions.DEFAULT);

                // 错误处理
                if (response.hasFailures())  {
                    processFailedItems(response, batch);
                } else {
                    successCount.addAndGet(batch.size());
                }
            } catch (IOException e) {
                log.error("ES 批量插入失败 [index: {}, batch size: {}]", index, batch.size(),  e);
                // 可添加重试逻辑
            }
        });

        log.info("ES 批量插入完成 [index: {}, total: {}, success: {}]", index, data.size(),  successCount.get());
    }

    /**
     * 处理失败记录
     */
    private static void processFailedItems(BulkResponse response, List<?> originalBatch) {
        for (BulkItemResponse item : response.getItems())  {
            if (item.isFailed())  {
                int itemIndex = item.getItemId();
                Object failedDoc = originalBatch.get(itemIndex);
                log.error("ES 插入失败 [id: {}, error: {}], 原始数据: {}",
                        item.getId(),
                        item.getFailureMessage(),
                        JSON.toJSONString(failedDoc));

                // 可添加失败记录重试或持久化
            }
        }
    }

    /**
     * 查询所有数据
     */
    public static <T> PageResult<T> queryDataAll(String index, Class<T> cls) {
        SearchRequest searchRequest = new SearchRequest(index);
        searchRequest.scroll(scroll);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.size(10000);
        searchRequest.source(searchSourceBuilder);
        // scroll滚动翻页
        List<SearchHit> searchHits = doScrollSearch(searchRequest);
        log.debug("elasticsearch数据查询sql:\n{}", searchSourceBuilder);

        List<T> resultData = searchHits.stream().map(it -> {
            JSONObject jsonObject = JSON.parseObject(it.getSourceAsString());
            jsonObject.put(ES_ID, it.getId());
            return jsonObject.toJavaObject(cls);
        }).collect(Collectors.toList());
        return new PageResult<>(resultData, Convert.toLong(resultData.size()));
    }

    private static List<SearchHit> doScrollSearch(SearchRequest searchRequest) {
        SearchResponse searchResponse = null;
        try {
            searchResponse = restHighLevelClient.search(searchRequest,  RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error(e.getMessage(),  e);
            throw new CustomException("elasticsearch 滚动查询失败");
        }

        String scrollId = Objects.requireNonNull(searchResponse).getScrollId();
        SearchHit[] searchHits = searchResponse.getHits().getHits();
        List<SearchHit> list = Arrays.stream(searchHits).collect(Collectors.toList());

        // 滚动查询核心逻辑
        List<SearchHit> result = doProcessScrollResults(scrollId, list);

        log.debug("elasticsearch 滚动查询结束···");
        // 清理滚动查询
        clearScroll(scrollId);
        return result;
    }

    /**
     * 处理滚动查询结果
     * @param initialScrollId 初始scrollId
     * @param initialList 初始结果列表
     * @return 包含最新scrollId和合并后结果列表的Pair
     */
    private static List<SearchHit> doProcessScrollResults(String initialScrollId, List<SearchHit> initialList) {
        String scrollId = initialScrollId;
        List<SearchHit> list = new ArrayList<>(initialList);
        SearchHit[] searchHits;
        SearchResponse searchResponse;

        do {
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            scrollRequest.scroll(scroll);
            try {
                searchResponse = restHighLevelClient.scroll(scrollRequest,  RequestOptions.DEFAULT);
            } catch (IOException e) {
                log.error(e.getMessage(),  e);
                break;
            }

            System.out.println("elasticsearch 正在滚动查询..." + scrollRequest);
            scrollId = searchResponse.getScrollId();
            searchHits = searchResponse.getHits().getHits();
            list.addAll(Arrays.asList(searchHits));
        } while (searchHits.length  > 0);
        return list;
    }

    /**
     * 清除scroll
     * @param scrollId 要清除的scrollId
     * @return 是否清除成功
     */
    private static boolean clearScroll(String scrollId) {
        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
        clearScrollRequest.addScrollId(scrollId);
        try {
            ClearScrollResponse clearScrollResponse = restHighLevelClient.clearScroll(clearScrollRequest,  RequestOptions.DEFAULT);
            return clearScrollResponse.isSucceeded();
        } catch (IOException e) {
            log.error(e.getMessage(),  e);
            throw new CustomException("elasticsearch 清除滚动查询失败");
        }
    }

    /**
     * 分页查询所有数据
     */
    public static PageResult<Map<String, Object>> queryDataPage(String index, SearchSourceBuilder sourceBuilder, String sortField, SortOrder sortOrder) {
        PageParam pageParam = PageUtil.getPageParam();
        Assert.notEmpty(index,INDEX_MSG);
        SearchRequest request = new SearchRequest(index);
        sourceBuilder.from(pageParam.getPageNum() - 1);
        sourceBuilder.size(pageParam.getPageSize());
        if (null != sortField) {
            sourceBuilder.sort(sortField, Optional.ofNullable(sortOrder).orElse(SortOrder.DESC));
        }
        sourceBuilder.trackTotalHits(Boolean.TRUE);
        request.source(sourceBuilder);
        try {
            log.debug("elasticsearch分页数据查询sql:\n{}", sourceBuilder);
            SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
            SearchHit[] hits = response.getHits().getHits();
            long count = response.getHits().getTotalHits().value;
            List<Map<String, Object>> list = Arrays.stream(hits).map(it -> {
                Map<String, Object> map = it.getSourceAsMap();
                // 仅限于我这里使用哦
                map.put(ES_ID, it.getId());
                return map;
            }).collect(Collectors.toList());
            return new PageResult<Map<String, Object>>(list,count);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        return new PageResult<>();
    }

    /**
     * 根据指定条件查询具体一条数据
     * @param index 索引
     * @param builder 构建条件
     * @return Map<String, Object>
     */
    public static Map<String, Object> queryOne(String index, SearchSourceBuilder builder,String sortField, SortOrder sortOrder) {
        Assert.notEmpty(index,INDEX_MSG);
        SearchRequest request = new SearchRequest(index);
        builder.size(1);
        builder.sort(sortField, Optional.ofNullable(sortOrder).orElse(SortOrder.DESC));
        request.source(builder);
        try {
            log.debug("elasticsearch单条数据查询sql:\n{}", builder);
            SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
            List<SearchHit> hits = Arrays.stream(response.getHits().getHits()).collect(Collectors.toList());
            if (CollUtil.isNotEmpty(hits)) {
                return hits.get(0).getSourceAsMap();
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        return new HashMap<>();
    }

    /**
     * 分页查询所有数据数量
     */
    public static long queryDataCount(String index, SearchSourceBuilder builder) {
        Assert.notEmpty(index,INDEX_MSG);
        CountRequest request = new CountRequest(index);
        builder.trackTotalHits(Boolean.TRUE);
        builder.fetchSource(false);
        request.source(builder);
        try {
            CountResponse countResponse = restHighLevelClient.count(request, RequestOptions.DEFAULT);
            return countResponse.getCount();
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        return 0;
    }

    /**
     * 根据id查询详细信息
     */
    public static <T> T queryDataById(String index, String id, Class<T> cls) {
        GetRequest getRequest = new GetRequest(index, id);
        try {
            GetResponse response = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
            if (response.isExists()) {
                JSONObject jsonObject = JSON.parseObject(response.getSourceAsString());
                return jsonObject.toJavaObject(cls);
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        return null;
    }

    /**
     * 修改文档数据
     */
    public static Boolean updateData(String index,String id,Object dyna) {
        UpdateRequest updateRequest = new UpdateRequest(index,id);
        JSONObject jsonObject = JSON.parseObject(JSON.toJSONString(dyna));
        updateRequest.doc(jsonObject,XContentType.JSON);
        try {
            UpdateResponse updateResponse = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
            DocWriteResponse.Result result = updateResponse.getResult();
            if (UPDATE.equalsIgnoreCase(result.getLowercase())) {
                return Boolean.TRUE;
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        return Boolean.FALSE;
    }

    /**
     * 删除文档数据
     */
    public static Boolean delData(String index,String id) {
        DeleteRequest deleteRequest = new DeleteRequest(index).id(id);
        try {
            DeleteResponse deleteResponse = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
            if (DELETE.equalsIgnoreCase(deleteResponse.getResult().getLowercase())) {
                return Boolean.TRUE;
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        return Boolean.FALSE;
    }

    /**
     * 分组统计
     * @param index 索引
     * @param field 分组字段
     * @param type 类型
     */
    public static List<Map<String, Object>> group(String index, String field, String type) {
        // 查询条件
        SearchRequest searchRequest = new SearchRequest(index);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
        boolQueryBuilder.must(QueryBuilders.matchAllQuery());
        searchSourceBuilder.query(boolQueryBuilder);
        // 查询0有利于加快查询速度
        searchSourceBuilder.size(0);
        searchRequest.source(searchSourceBuilder);

        // 分组; time根据时间分组，默认根据text分组
        TermsAggregationBuilder aggregationBuilder = AggregationBuilders
                // 分组名称
                .terms("group_name");
        aggregationBuilder.field(field + ".keyword");
/*        // 作用是当统计的组中数量小于minDocCount定义的值，将不会出现在统计结果中
        aggregationBuilder.minDocCount(1);*/
        if (type.equals("time")) {
            aggregationBuilder.format("HH:mm");
        }

        // 将聚合添加到搜索源构建器
        searchSourceBuilder.aggregation(aggregationBuilder);

        try {
            SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            ParsedStringTerms parsedStringTerms = response.getAggregations().get("group_name");
            List<? extends Terms.Bucket> buckets = parsedStringTerms.getBuckets();
            return Arrays.stream(buckets.toArray()).map(BeanUtil::beanToMap).collect(Collectors.toList());
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        return null;
    }

    /**
     * 查询最大值和最小值
     * @param index 索引
     * @param field 统计字段
     * @param type true最大值，false最小值
     * @return 值
     */
    public static Double stats(String index, String field, Boolean type) {
        // 查询条件
        SearchRequest searchRequest = new SearchRequest(index);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
        boolQueryBuilder.must(QueryBuilders.matchAllQuery());
        searchSourceBuilder.query(boolQueryBuilder);
        // 查询0有利于加快查询速度
        searchSourceBuilder.size(0);
        searchRequest.source(searchSourceBuilder);

        SearchResponse response;
        try {
            if (type) {
                // 该方式只用于求所有数据中某个字段的最大值和最小值
                MaxAggregationBuilder aggregationBuilder = AggregationBuilders.max("max_stats_score").field(field);
                searchSourceBuilder.aggregation(aggregationBuilder);
                response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
                ParsedMax maxStatsScore = response.getAggregations().get("max_stats_score");
                return maxStatsScore.getValue();
            }else {
                MinAggregationBuilder aggregationBuilder = AggregationBuilders.min("min_stats_score").field(field);
                searchSourceBuilder.aggregation(aggregationBuilder);
                response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
                ParsedMin minStatsScore = response.getAggregations().get("min_stats_score");
                return minStatsScore.getValue();
            }

        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        return null;
    }

    /**
     * 统计分组中数量、最大值和最小值
     * @param index 索引
     * @param groupField 分组字段
     * @param staticField 统计字段
     * @param type true最大值，false最小值
     */
    public static List<GroupResp> getGroupStats(String index, String groupField, String staticField, Boolean type, String sort) {
        List<GroupResp> result = new ArrayList<>();
        // 查询条件
        SearchRequest searchRequest = new SearchRequest(index);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
        boolQueryBuilder.must(QueryBuilders.matchAllQuery());
        searchSourceBuilder.query(boolQueryBuilder);
        // 查询0有利于加快查询速度
        searchSourceBuilder.size(0);
        searchRequest.source(searchSourceBuilder);

        SearchResponse response;
        BucketOrder bucketOrder = BucketOrder.count("acs".equals(sort));
        try {
            if (type) {
                TermsAggregationBuilder aggregationBuilder = getTermsAggregationBuilder("max", groupField, bucketOrder);
                // 分段统计
                String maxGroup = "max_" + staticField + "group";
                aggregationBuilder.subAggregation(AggregationBuilders
                        // "max_"+ staticField +"group" 最大值名称
                        .max(maxGroup)
                        // 统计字段
                        .field(staticField));
                searchSourceBuilder.aggregation(aggregationBuilder);
                response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
                ParsedLongTerms parsedLongTerms = response.getAggregations().get("max_" + groupField);
                parsedLongTerms.getBuckets().forEach(it-> toBean(result, it, maxGroup,type));
            }else {
                TermsAggregationBuilder aggregationBuilder = getTermsAggregationBuilder("min", groupField, bucketOrder);
                // 分段统计
                String minGroup = "min_" + staticField + "group";
                aggregationBuilder.subAggregation(AggregationBuilders
                        // "min_"+ staticField +"group" 最小值名称
                        .min(minGroup)
                        .field(staticField));
                searchSourceBuilder.aggregation(aggregationBuilder);
                response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
                ParsedLongTerms parsedLongTerms = response.getAggregations().get("min_" + groupField);
                parsedLongTerms.getBuckets().forEach(it-> toBean(result, it, minGroup,type));
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        return result;
    }

    private static void toBean(List<GroupResp> result, Terms.Bucket it, String name, Boolean type) {
        GroupResp groupResp = new GroupResp();
        groupResp.setKeyAsString(it.getKeyAsString());
        groupResp.setDocCount(it.getDocCount());
        if (type) {
            ParsedMax parsedMax = it.getAggregations().get(name);
            groupResp.setMaxScore(parsedMax.getValue());
        }else {
            ParsedMin parsedMin = it.getAggregations().get(name);
            groupResp.setMaxScore(parsedMin.getValue());
        }

        result.add(groupResp);
    }

    private static TermsAggregationBuilder getTermsAggregationBuilder(String type, String groupField, BucketOrder bucketOrder) {
        TermsAggregationBuilder aggregationBuilder = AggregationBuilders
                // 分组名称
                .terms(type + "_" + groupField)
                // 分组字段
                .field(groupField);
        aggregationBuilder.order(bucketOrder);
        return aggregationBuilder;
    }

}

